package com.shujia.hbaseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 思考：
 * 1、如何将所有的学生信息查询出来？有几种方式？
 * 2、加入我给一个条件查询，你会不会呢？比如说：查姓王，年龄24以下的，理科的，女生
 * 3、我们今天学习了预分region，如何用代码去创建呢？
 * 4、如何设置过期时间呢？
 * 5、hbase的读写流程是什么样子呢？你可以自己画出来吗？
 * 6、hbase中涉及到的数据结构有哪些？
 * 7、二级索引了解一下
 * 8、如何通过代码实现二级索引，有什么好处？
 */

public class ClintToHbase {
    private HConnection conn;
    private HBaseAdmin hAdmin;

    @Before
    public void connect() {
        try {
            //1、获取Hadoop的相关配置环境
            Configuration conf = new Configuration();

            //2、获取zookeeper的配置
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
            //获取与Hbase的连接，这个连接是将来可以用户获取hbase表的
            conn = HConnectionManager.createConnection(conf);

            //将来我们要对表做DDL相关操作，而对表的操作在hbase架构中是有HMaster
            hAdmin = new HBaseAdmin(conf);

            System.out.println("建立连接成功:" + conn + ", HMaster获取成功：" + hAdmin);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 在HBase中创建表
     * <p>
     * create 'students','info'
     */
    @Test
    public void createTable() {
        try {
            //使用HTableDescriptor类创建一个表对象
            HTableDescriptor students = new HTableDescriptor("students");

            //在创建表的时候，至少指定一个列簇
            HColumnDescriptor info = new HColumnDescriptor("info");
            info.setBloomFilterType(BloomType.ROW);


            //将列簇添加到表中
            students.addFamily(info);
            //真正的执行，是由HMaster
            //hAdmin
            hAdmin.createTable(students);
            System.out.println(Bytes.toString(students.getName()) + "表 创建成功。。。");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 在还没有说预分region之前，插入数据的时候，所有的数据都会被集中的插入到一个region中，当前只有一个regionserver管理这个region
     * 会导致热点问题，采用预分region解决热点问题
     * 如何创建region表
     *
     */
    @Test
    public void createPrepareRegionsTable(){

        try {
            //建表之前判断一下表是否存在
            boolean b = hAdmin.tableExists("test1");
            if(b){
                hAdmin.disableTable("test1");
                hAdmin.deleteTable("test1");
            }

            HTableDescriptor test1 = new HTableDescriptor("test1");
            HColumnDescriptor info = new HColumnDescriptor("info");
            //对一个列簇设置过期时间
            info.setTimeToLive(60*60*24);
            //对一个列簇设置版本号
            info.setMaxVersions(3);
            test1.addFamily(info);

            byte[][] splitkeys = {
                    "1001".getBytes(),
                    "2001".getBytes()
            };
            hAdmin.createTable(test1,splitkeys);
            System.out.println("预分region表创建完毕！！！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 删除Hbase中的表
     */
    @Test
    public void deleteTable() {
        //1、判断一下要删除的表是否存在
        try {
            if (hAdmin.tableExists("students")) {
                //如果表存在，就可以去删除
                //直接删除的话，是删除不了的
                //先禁用表
                hAdmin.disableTable("students");
                hAdmin.deleteTable("students");
                System.out.println(conn.getTable("students").getName() + "表 成功被删除....");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 往表中添加一条数据
     * <p>
     * put 'students','1001','info:name','xiaohu'
     */
    @Test
    public void putData() {
        try {
            //1、获取到表的实例
            HTableInterface students = conn.getTable("students");

            //创建一个Put实例，并且给一个行键（rowkey）
            Put put = new Put("1001".getBytes());

            //将这个行键对应的值，添加列簇，列，具体的值
            put.add("info".getBytes(), "name".getBytes(), "xiaohu".getBytes());

            //表的实例将数据添加到表中
            students.put(put);

            String s = put.toString();
            System.out.println(s);

            System.out.println(students.getName() + "表 成功插入一条数据");

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取hbase中的一条数据
     * <p>
     * get 'students','1001'
     */
    @Test
    public void getData() {
        try {
            //1、获取到表的实例
            HTableInterface students = conn.getTable("students");
            //获取get对象，通过rowkey获取
            Get get = new Get("1001".getBytes());
            //获取到了行键对应的信息
            String rowkey = Bytes.toString(get.getRow());
//            System.out.println(s);

            //表的实例调用get方法，返回的是一个结果集
            Result result = students.get(get);

            //调用getValue()方法获取到一个字节数组中的一个结果集
            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            System.out.println("查询结束，查询结果如下：");
            System.out.println(rowkey + ":" + name);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一次性往表中添加多条数据
     * <p>
     * students.txt
     */
    @Test
    public void putAll() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");
            //创建一个List集合对象
            ArrayList<Put> puts = new ArrayList<>();

            //读取数据 --- IO
            BufferedReader br = new BufferedReader(new FileReader("E:\\projects\\IdeaProjects\\hadoop-bigdata17\\hadoop-hbase\\data\\students.txt"));

            String line = null;
            while ((line = br.readLine()) != null) {
                String[] split = line.split(",");
                String id = split[0];
                String name = split[1];
                String age = split[2];
                String gender = split[3];
                String clazz = split[4];

                //把每一行组成一个put对象
                Put put = new Put(id.getBytes());

                //为一行添加多列

                put.add("info".getBytes(), "name".getBytes(), name.getBytes());
                put.add("info".getBytes(), "age".getBytes(), age.getBytes());
                put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
                put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

                //每一行组成一个put对象后添加到集合中
                //put 'students',id,'info:name',name
                //put 'students',id,'info:age',age
                //put 'students',id,'info:gender',gender
                //put 'students',id,'info:clazz',clazz
                puts.add(put);
            }

            //调用表的put方法，将集合数据添加到表中
            students.put(puts);
            System.out.println("学生信息表添加完毕！！");

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 查询结果
     */
    @Test
    public void getData2() {
        try {
            //获取到表的实例
            HTableInterface students = conn.getTable("students");

            //创建一个Get实例，创建的时候传入一个rowkey
            Get get = new Get("1500100018".getBytes());
            String id = Bytes.toString(get.getRow());

            //获取到结果集
            Result result = students.get(get);

            /**
             *  当我们知道列的个数和列名的时候
             */
            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

            System.out.println("-------------------------------------------------------------------");
            /**
             *  在不知道列的数量和列名的时候，可以获取一行中的所有单元格cell
             */
            List<Cell> list = result.listCells();
            for (Cell cell : list) {
//                String s = Bytes.toString(cell.getValue());
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String s = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(col + ":" + s);
                System.out.println("\t");
            }
            System.out.println();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 一次获取多条数据
     * scan 'students'
     */
    @Test
    public void ScanAll() {
        try {
            //获取到表的实例
            HTableInterface students = conn.getTable("students");
            //创建的Scan的实例
            Scan scan = new Scan();

            //通常情况下，如果就仅仅查看一下数据的样式，没有必要获取所有的数据展示
            //需要定义一下查询的范围，含头不含尾的[start,stop)
            //传入的是rowkey的值
            scan.setStartRow("1500100001".getBytes());
            scan.setStopRow("1500100011".getBytes());

            //将scan加入到表中
            ResultScanner scanner = students.getScanner(scan);
            //调用ResultScanner中的next()方法，next()是指向下一个数据的索引。有点类似于我之前迭代器的指针
            //用while循环遍历
            Result rs = null;
            while ((rs = scanner.next()) != null) {
                String id = Bytes.toString(rs.getRow());
//                List<Cell> cells = rs.listCells();
//                System.out.print("id:" + id);
//                System.out.print("\t");
//                for (Cell cell : cells) {
////                String s = Bytes.toString(cell.getValue());
//                    String col = Bytes.toString(CellUtil.cloneQualifier(cell));
//                    String s = Bytes.toString(CellUtil.cloneValue(cell));
//                    System.out.print(col + ":" + s);
//                    System.out.print("\t");
//                }
//                System.out.println();

                String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
                String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
                String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
                String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
                System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

            }


        } catch (IOException e) {
            e.printStackTrace();
        }

    }


    @After
    public void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("conn连接已经关闭.....");
        }

        if (hAdmin != null) {
            try {
                hAdmin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("HMaster已经关闭......");
        }
    }
}
